/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.impl.consumer;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;

import com.alibaba.rocketmq.client.VirtualEnvUtil;
import com.alibaba.rocketmq.client.consumer.PullCallback;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.client.consumer.PullStatus;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.hook.FilterMessageContext;
import com.alibaba.rocketmq.client.hook.FilterMessageHook;
import com.alibaba.rocketmq.client.impl.CommunicationMode;
import com.alibaba.rocketmq.client.impl.FindBrokerResult;
import com.alibaba.rocketmq.client.impl.factory.MQClientInstance;
import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.message.MessageAccessor;
import com.alibaba.rocketmq.common.message.MessageConst;
import com.alibaba.rocketmq.common.message.MessageDecoder;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.header.PullMessageRequestHeader;
import com.alibaba.rocketmq.common.protocol.heartbeat.SubscriptionData;
import com.alibaba.rocketmq.common.protocol.route.TopicRouteData;
import com.alibaba.rocketmq.common.sysflag.PullSysFlag;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

/**
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-24
 */
public class PullAPIWrapper {
	private final Logger log = ClientLogger.getLog();
	private ConcurrentHashMap<MessageQueue, AtomicLong/* brokerId */> pullFromWhichNodeTable = new ConcurrentHashMap<MessageQueue, AtomicLong>(32);

	private final MQClientInstance mQClientFactory;
	private final String consumerGroup;
	private final boolean unitMode;

	private volatile boolean connectBrokerByUser = false;
	private volatile long defaultBrokerId = MixAll.MASTER_ID;

	public PullAPIWrapper(MQClientInstance mQClientFactory, String consumerGroup, boolean unitMode) {
		this.mQClientFactory = mQClientFactory;
		this.consumerGroup = consumerGroup;
		this.unitMode = unitMode;
	}

	public void updatePullFromWhichNode(final MessageQueue mq, final long brokerId) {
		AtomicLong suggest = this.pullFromWhichNodeTable.get(mq);
		if (null == suggest) {
			// 若以messageQueue为key值从pullFromWhichNodeTable中获取的BrokerId为空
			// 则将PullResultExt.suggestWhichBrokerId存入该列表中
			this.pullFromWhichNodeTable.put(mq, new AtomicLong(brokerId));
		} else {
			// 否则更新该MessageQueue对应的value值为suggestWhichBrokerId
			suggest.set(brokerId);
		}
	}

	private Random random = new Random(System.currentTimeMillis());

	public int randomNum() {
		int value = random.nextInt();
		if (value < 0) {
			value = Math.abs(value);
			if (value < 0)
				value = 0;
		}
		return value;
	}

	private String computPullFromWhichFilterServer(final String topic, final String brokerAddr) throws MQClientException {
		// 5.1)根据topic参数值从MQClientInstance.topicRouteTable: ConcurrentHashMap<String/*Topic*/,TopicRouteData>变量中获取TopicRouteData对象
		ConcurrentHashMap<String, TopicRouteData> topicRouteTable = this.mQClientFactory.getTopicRouteTable();
		if (topicRouteTable != null) {
			TopicRouteData topicRouteData = topicRouteTable.get(topic);
			// 5.2)以Broker地址为参数从该TopicRouteData对象的filterServerTable:HashMap<String/* brokerAddr*/,List<String>/* FilterServer*/>变量中获取该Broker下面的所有Filter服务器地址列表；
			List<String> list = topicRouteData.getFilterServerTable().get(brokerAddr);

			if (list != null && !list.isEmpty()) {
				// 5.3)若该地址列表不为空，则随机选择一个Filter服务器地址返回；
				return list.get(randomNum() % list.size());
			}
		}
		// 否则向调用层抛出异常，该pullKernelImpl方法结束；
		throw new MQClientException("Find Filter Server Failed, Broker Addr: " + brokerAddr + " topic: " + topic, null);
	}

	public PullResult processPullResult(final MessageQueue mq, final PullResult pullResult, final SubscriptionData subscriptionData) {
		// 1.6）向NameServer发送GET_KV_CONFIG请求码获取NAMESPACE_PROJECT_CONFIG和本地IP下面的value值，赋值给projectGroupPrefix变量
		final String projectGroupPrefix = this.mQClientFactory.getMQClientAPIImpl().getProjectGroupPrefix();
		//System.out.println("projectGroupPrefix=======================pullAPIWrapper processPullResult===" + projectGroupPrefix);
		PullResultExt pullResultExt = (PullResultExt) pullResult;

		// 1.1）调用PullAPIWrapper.updatePullFromWhichNode(MessageQueue mq, long brokerId)方法
		// 用Broker返回的PullResultExt.suggestWhichBrokerId变量值更新PullAPIWrapper.pullFromWhichNodeTable:ConcurrentHashMap <MessageQueue,AtomicLong/* brokerId */>变量中当前拉取消息PullRequest.messageQueue对象对应的BrokerId。
		// 若以messageQueue为key值从pullFromWhichNodeTable中获取的BrokerId为空
		// 则将PullResultExt.suggestWhichBrokerId存入该列表中，
		// 否则更新该MessageQueue对应的value值为suggestWhichBrokerId；
		this.updatePullFromWhichNode(mq, pullResultExt.getSuggestWhichBrokerId());

		// 1.2）若pullResult.status=FOUND，则继续下面的处理逻辑，
		// 否则设置PullResultExt.messageBinary=null并返回该PullResult对象；
		if (PullStatus.FOUND == pullResult.getPullStatus()) {
			ByteBuffer byteBuffer = ByteBuffer.wrap(pullResultExt.getMessageBinary());
			// 1.3）对PullResultExt.messageBinary变量进行解码，得到MessageExt列表；
			List<MessageExt> msgList = MessageDecoder.decodes(byteBuffer);

			// 1.4) Consumer端消息过滤。
			List<MessageExt> msgListFilterAgain = msgList;
			// 若SubscriptionData.tagsSet集合（在5.5.1小节中拉取消息之前以topic获取的订阅关系数据）不为空并且SubscriptionData. classFilterMode为false（在初始化DefaultMQPushConsumer时可以设置这两个值）
			if (!subscriptionData.getTagsSet().isEmpty() && !subscriptionData.isClassFilterMode()) {
				msgListFilterAgain = new ArrayList<MessageExt>(msgList.size());
				for (MessageExt msg : msgList) {
					if (msg.getTags() != null) {
						// 则遍历MessageExt列表，检查每个MessageExt对象的tags值（在commitlog数据的properties字段的"TAGS"属性值）是否在SubscriptionData.tagsSet集合中
						if (subscriptionData.getTagsSet().contains(msg.getTags())) {
							// 只保留MessageExt.tags此tagsSet集合中的MessageExt对象，构成新的MessageExt列表，取名msgListFilterAgain
							// 否则新的列表msgListFilterAgain等于上一步的MessageExt列表；
							msgListFilterAgain.add(msg);
						}
					}
				}
			}

			/**
			 *  1.5）检查PullAPIWrapper.filterMessageHookList列表是否为空
			 * （可在应用层通过DefaultMQPullConsumerImpl.registerFilterMessageHook (FilterMessageHook hook)方法设置），
			 * 若不为空则调用该列表中的每个FilterMessageHook对象的filterMessage方法；
			 * 由应用层实现FilterMessageHook接口的filterMessage方法，可以在该方法中对消息再次过滤；
			 */
			if (this.hasHook()) {
				FilterMessageContext filterMessageContext = new FilterMessageContext();
				filterMessageContext.setUnitMode(unitMode);
				filterMessageContext.setMsgList(msgListFilterAgain);
				this.executeHook(filterMessageContext);
			}

			if (!UtilAll.isBlank(projectGroupPrefix)) {
				// 若不为空则
				// 除了将PullResult.minOffset和PullResult.maxOffset值设置到每个MessageExt对象的properties属性中之外，
				// 还从projectGroupPrefix变量值开头的topic中去掉projectGroupPrefix值部分，
				// 然后将新的topic设置到MessageQueue、SubscriptionData的topic以及每个MessageExt对象的topic变量；
				subscriptionData.setTopic(VirtualEnvUtil.clearProjectGroup(subscriptionData.getTopic(), projectGroupPrefix));
				mq.setTopic(VirtualEnvUtil.clearProjectGroup(mq.getTopic(), projectGroupPrefix));
				for (MessageExt msg : msgListFilterAgain) {
					msg.setTopic(VirtualEnvUtil.clearProjectGroup(msg.getTopic(), projectGroupPrefix));

					MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MIN_OFFSET, Long.toString(pullResult.getMinOffset()));
					MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MAX_OFFSET, Long.toString(pullResult.getMaxOffset()));
				}
			} else {
				// 若该值为空则将PullResult.minOffset和PullResult.maxOffset值设置到每个MessageExt对象的properties属性中，
				// 其中属性名称分别为"MIN_OFFSET"和"MAX_OFFSET"
				for (MessageExt msg : msgListFilterAgain) {
					MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MIN_OFFSET, Long.toString(pullResult.getMinOffset()));
					MessageAccessor.putProperty(msg, MessageConst.PROPERTY_MAX_OFFSET, Long.toString(pullResult.getMaxOffset()));
				}
			}

			// 1.7）将新组建的MessageExt列表msgListFilterAgain赋值给PullResult.msgFoundList变量；
			pullResultExt.setMsgFoundList(msgListFilterAgain);
		}

		// 1.8）设置PullResultExt.messageBinary=null，并返回该PullResult对象；
		pullResultExt.setMessageBinary(null);

		return pullResult;
	}

	public long recalculatePullFromWhichNode(final MessageQueue mq) {
		// 先判断PullAPIWrapper.connectBrokerByUser变量是否为true（在FiltersrvController中启动时会设置为true，默认为false）
		if (this.isConnectBrokerByUser()) {
			// 若是则直接返回0（主用Broker的brokerId）
			return this.defaultBrokerId;
		}

		// 否则以MessageQueue对象为参数从PullAPIWrapper.pullFromWhichNodeTable:ConcurrentHashMap<MessageQueue, AtomicLong/* brokerId */>获取brokerId
		AtomicLong suggest = this.pullFromWhichNodeTable.get(mq);
		if (suggest != null) {
			// 若该值不为null则返回该值，
			return suggest.get();
		}

		// 否则返回0（主用Broker的brokerId）
		return MixAll.MASTER_ID;
	}

	public PullResult pullKernelImpl(//
			final MessageQueue mq, // 1
			final String subExpression, // 2
			final long subVersion, // 3
			final long offset, // 4
			final int maxNums, // 5
			final int sysFlag, // 6
			final long commitOffset, // 7
			final long brokerSuspendMaxTimeMillis, // 8
			final long timeoutMillis, // 9
			final CommunicationMode communicationMode, // 10
			final PullCallback pullCallback// 11
	) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		// 2、调用MQClientInstance.findBrokerAddressInSubscribe(String brokerName ,long brokerId,boolean onlyThisBroker) 方法查找Broker地址，
		// 其中onlyThisBroker=false，表示若指定的brokerId未获取到地址则获取其他BrokerId的地址也行。
		// 在该方法中根据brokerName和brokerId参数从MQClientInstance.brokerAddrTable: ConcurrentHashMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>>变量中获取对应的Broker地址，
		// 若获取不到则从brokerName下面的Map列表中找其他地址返回即可；
		FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), //
				// 1、获取Broker的ID。
				// 以入参MessageQueue对象为参数调用PullAPIWrapper.recalculatePullFromWhichNode(MessageQueue mq)方法，
				// 在该方法中，先判断PullAPIWrapper.connectBrokerByUser变量是否为true（在FiltersrvController中启动时会设置为true，默认为false），
				// 若是则直接返回0（主用Broker的brokerId）；
				// 否则以MessageQueue对象为参数从PullAPIWrapper.pullFromWhichNodeTable:ConcurrentHashMap<MessageQueue, AtomicLong/* brokerId */>获取brokerId，
				// 若该值不为null则返回该值，否则返回0（主用Broker的brokerId）；
				this.recalculatePullFromWhichNode(mq), //
				false // onlyThisBroker=false，表示若指定的brokerId未获取到地址则获取其他BrokerId的地址也行
		);
		if (null == findBrokerResult) {
			// 3、若在上一步未获取到Broker地址，则以topic参数调用MQClientInstance.updateTopicRouteInfoFromNameServer(String topic)方法，
			// 然后在执行第2步的操作，直到获取到Broker地址为止；
			this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
			findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), this.recalculatePullFromWhichNode(mq), false);
		}

		if (findBrokerResult != null) {
			int sysFlagInner = sysFlag;

			// 4、若获取的Broker地址是备用Broker，则将标记位sysFlag的第1个字节置为0，即在消费完之后不提交消费进度；
			if (findBrokerResult.isSlave()) {
				sysFlagInner = PullSysFlag.clearCommitOffsetFlag(sysFlagInner);
			}

			// 6、构建PullMessageRequestHeader对象，其中queueOffset变量值等于入参offset；
			PullMessageRequestHeader requestHeader = new PullMessageRequestHeader();
			requestHeader.setConsumerGroup(this.consumerGroup);
			requestHeader.setTopic(mq.getTopic());
			requestHeader.setQueueId(mq.getQueueId());
			requestHeader.setQueueOffset(offset);
			requestHeader.setMaxMsgNums(maxNums);
			requestHeader.setSysFlag(sysFlagInner);
			requestHeader.setCommitOffset(commitOffset);
			requestHeader.setSuspendTimeoutMillis(brokerSuspendMaxTimeMillis);
			requestHeader.setSubscription(subExpression);
			requestHeader.setSubVersion(subVersion);

			String brokerAddr = findBrokerResult.getBrokerAddr();
			if (PullSysFlag.hasClassFilterFlag(sysFlagInner)) {
				// 5、检查标记位sysFlag的第4个字节（即SubscriptionData. classFilterMode）是否为1；
				// 若等于1，则调用PullAPIWrapper.computPullFromWhichFilterServer(String topic, String brokerAddr)方法获取Filter服务器地址
				brokerAddr = computPullFromWhichFilterServer(mq.getTopic(), brokerAddr);
			}

			// ******************发送远程请求拉取消息的逻辑（PULL_MESSAGE）******************
			// 7、若执行了第5步则向获取的Filter服务器发送PULL_MESSAGE请求信息，否则向Broker发送PULL_MESSAGE请求信息。
			PullResult pullResult = this.mQClientFactory.getMQClientAPIImpl().pullMessage(//
					brokerAddr, //
					requestHeader, //
					timeoutMillis, //
					communicationMode, //
					pullCallback);

			return pullResult;
		}

		throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
	}

	private ArrayList<FilterMessageHook> filterMessageHookList = new ArrayList<FilterMessageHook>();

	public boolean hasHook() {
		return !this.filterMessageHookList.isEmpty();
	}

	public void registerFilterMessageHook(ArrayList<FilterMessageHook> filterMessageHookList) {
		this.filterMessageHookList = filterMessageHookList;
	}

	public void executeHook(final FilterMessageContext context) {
		if (!this.filterMessageHookList.isEmpty()) {
			for (FilterMessageHook hook : this.filterMessageHookList) {
				try {
					hook.filterMessage(context);
				} catch (Throwable e) {
					log.error("execute hook error. hookName={}", hook.hookName());
				}
			}
		}
	}

	public long getDefaultBrokerId() {
		return defaultBrokerId;
	}

	public void setDefaultBrokerId(long defaultBrokerId) {
		this.defaultBrokerId = defaultBrokerId;
	}

	public boolean isConnectBrokerByUser() {
		return connectBrokerByUser;
	}

	public void setConnectBrokerByUser(boolean connectBrokerByUser) {
		this.connectBrokerByUser = connectBrokerByUser;

	}
}
